package com.sui.bigdata.flink.sql.side.program.util;

import com.google.common.hash.BloomFilter;
import com.google.common.hash.Funnels;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;

/**
 * @author YongChen
 * @date 2020/6/1 13:53
 * @description
 * @email yong_chen@sui.com
 */
public  class MyProcessWindowFunction extends ProcessWindowFunction<Tuple3<String, Integer,Integer>, Tuple3<String,Long, Integer>, Tuple, TimeWindow> {
    private static final long serialVersionUID = 1L;
    private static final Logger LOGGER = LoggerFactory.getLogger(MyProcessWindowFunction.class);
    private static final int BF_CARDINAL_THRESHOLD = 1000000;
    private static final double BF_FALSE_POSITIVE_RATE = 0.01;

    private transient ValueState<BloomFilter<String>> subOrderFilter;

    @Override
    public void open(Configuration parameters) throws Exception {
        ValueStateDescriptor<BloomFilter<String>> subOrderFilterDescriptor = new ValueStateDescriptor<BloomFilter<String>>("boom_filter", TypeInformation.of(new TypeHint<BloomFilter<String>>() {
        }));
        subOrderFilter = getRuntimeContext().getState(subOrderFilterDescriptor);
    }

    @Override
    public void clear(Context context) throws Exception {
        super.clear(context);
        subOrderFilter.update(BloomFilter.create(Funnels.stringFunnel(Charset.forName("GBK")), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE));
    }

    @Override
    public void process(Tuple key, Context context, Iterable<Tuple3<String, Integer, Integer>> elements, Collector< Tuple3<String,Long, Integer>> out) throws Exception {
        BloomFilter<String> bloomFilter = subOrderFilter.value();
        if (null == bloomFilter) {
            bloomFilter = BloomFilter.create(Funnels.stringFunnel(Charset.forName("GBK")), BF_CARDINAL_THRESHOLD, BF_FALSE_POSITIVE_RATE);
        }

        int sum = 0;
        for (Tuple3<String, Integer, Integer> tuple3 : elements) {
            String subOrderId = tuple3.f0;
            if (!bloomFilter.mightContain(subOrderId)) {
                bloomFilter.put(subOrderId);
                subOrderFilter.update(bloomFilter);
                sum += 1;
            }
        }

        out.collect(new Tuple3<String,Long, Integer>(key.getField(0), context.window().getStart(),sum));

    }
}
